package cc.chinagps.gateway.mq.writer;

import cc.chinagps.gateway.cache.DataCache;
import cc.chinagps.gateway.cache.ICache;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import java.util.List;

@Data
@Slf4j
public abstract class AbstractWriter<T> extends Thread implements IWriter<T> {
    private ICache cache;
    protected String writerName;
    protected int cacheSize;
    private List<T> tempGet;
    private long totalDataSize = 0;
    protected Session session;
    protected MessageProducer producer;
    protected int type;
    protected int mode;

    public AbstractWriter(Session session, String name, int type, int mode, int cacheSize) {
        this.session = session;
        this.writerName = name;
        this.type = type;
        this.mode = mode;
        this.cacheSize = cacheSize;
    }

    public void init() throws Exception {
        Destination destination;
        if (type == 1) {
            destination = session.createQueue(writerName);
        } else {
            destination = session.createTopic(writerName);
        }
        producer = session.createProducer(destination);
        if (mode == 1) {
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        } else {
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
        }
        cache = new DataCache(writerName, cacheSize);
    }

    @Override
    public void startService() {
        try {
            init();
            start();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }

    }

    @Override
    public void addData(T data) {
        cache.addData(data);
    }

    @Override
    public void run() {
        while (true) {
            try {
                save();
            } catch (Throwable e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    private void save() {
        tempGet = cache.getData();
        try {
            saveDataToMQ(tempGet);
            totalDataSize += tempGet.size();
        } catch (Throwable e) {
            log.error(e.getMessage(), e);
        }
        tempGet.clear();
    }

    protected abstract void saveDataToMQ(List<T> list) throws Exception;
}
